//package org.apache.flink.state
//
//import java.util
//import java.util.Collections
//
//import org.apache.flink.api.common.functions.RichFlatMapFunction
//import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, MapState, MapStateDescriptor, ValueState, ValueStateDescriptor}
//import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
//import org.apache.flink.api.scala._
//import org.apache.flink.configuration.Configuration
//import org.apache.flink.shaded.guava18.com.google.common.collect.Lists
//import org.apache.flink.util.Collector
//
//import scala.collection.JavaConversions._
//
///**
// * 需求：当接收到相同key的元素个数=3，就计算元素的value的平均值
// *
// *
// * TODO... 请使用MapState实现相同的功能
// **/
//object KeyedStateApp {
//  def main(args: Array[String]): Unit = {
//    val env = StreamExecutionEnvironment.getExecutionEnvironment
//    env.fromCollection(List(
//      (1L, 3L),
//      (1L, 7L),
//      (2L, 4L),
//      (1L, 5L),
//      (2L, 2L),
//      (2L, 5L)
//    )).keyBy(_._1)
//        .flatMap(new AvgWithValueState)
//        .print()
//
//    env.execute(getClass.getCanonicalName)
//  }
//}
//
//
//
//class AvgWithListState extends RichFlatMapFunction[(Long,Long), (Long, Double)] {
//
//  private var listState: ListState[(Long, Long)] = _
//
//  override def open(parameters: Configuration): Unit = {
//    listState = getRuntimeContext.getListState(
//      new ListStateDescriptor[(Long, Long)]("average", createTypeInformation[(Long, Long)])
//    )
//  }
//
//  override def flatMap(value: (Long, Long), out: Collector[(Long, Double)]): Unit = {
//    val currentState = listState.get()
//
//    if(currentState == null) {
//      listState.addAll(Collections.emptyList())
//    }
//
//    listState.add(value)
//
//    val elements: util.ArrayList[(Long, Long)] = Lists.newArrayList(listState.get())
//
//    if(elements.size() == 3) {
//      var count = 0L
//      var sum = 0L
//
//      for(ele <- elements) {
//        count += 1
//        sum += ele._2
//      }
//
//      val avg = sum / count.toDouble
//      out.collect(value._1, avg)
//
//      listState.clear()
//    }
//
//  }
//}
//
//class AvgWithValueState extends RichFlatMapFunction[(Long,Long), (Long, Double)] {
//
//  private var valueState: ValueState[(Long, Long)] = _
//
//  // 从上下文中根据名称获取到state
//  override def open(parameters: Configuration): Unit = {
//    valueState = getRuntimeContext.getState(
//      new ValueStateDescriptor[(Long, Long)]("average", createTypeInformation[(Long, Long)])
//    )
//  }
//
//  override def flatMap(input: (Long, Long), out: Collector[(Long, Double)]): Unit = {
//    // 获取state
//    val tmpCurrentSum = valueState.value()
//
//    val currentSum = if(tmpCurrentSum != null) {
//      tmpCurrentSum
//    } else {
//      (0L,0L)
//    }
//
//    // 平均数 = 和 / 次数
//    val newSum = (currentSum._1 + 1, currentSum._2 + input._2)
//    valueState.update(newSum)
//
//    if(newSum._1 >= 3) {
//      out.collect((input._1, newSum._2/newSum._1.toDouble))
//      valueState.clear()
//    }
//
//  }
//}
